package com.shujia.compute

import java.sql.DriverManager
import java.text.SimpleDateFormat

import com.shujia.util.{HbaseUtil, SparkTool}
import org.apache.hadoop.hbase.client.Put
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.streaming.kafka.KafkaUtils

import scala.collection.mutable.ListBuffer

object TrafficMonitorApp extends SparkTool {
  /**
    * 在run方法里面编写spark业务逻辑
    */
  override def run(args: Array[String]): Unit = {
    /**
      * 稽查布控
      *
      */

    val ssc = new StreamingContext(sc, Durations.seconds(5))


    val topics = Map("car" -> 2)

    //消费kafka数据
    val cars = KafkaUtils.createStream(
      ssc,
      "node1:2181,node2:2181,node3:2181",
      "asdsadas",
      topics
    )

    /**
      * 读取mysql中布控列表
      * 和实时数据进行匹配 如果匹配成功加车辆出现的时间和位置保存到mysql中
      *
      */


    cars
      .map(_._2)
      .foreachRDD(rdd => {

        //创建mysql连接 构建广播变量

        Class.forName("com.mysql.jdbc.Driver")

        val con = DriverManager.getConnection("jdbc:mysql://node1:3306/car", "root", "123456")

        val sql = "select * from t_monitor"

        val stat = con.prepareStatement(sql)
        val rs = stat.executeQuery()

        val cars = new ListBuffer[String]

        while (rs.next()) {
          val carId = rs.getString("car_id")
          cars.+=(carId)
        }

        println("正在布控的车辆" + cars)
        /**
          * 每一个batch都会更新一次
          *
          */
        val broCars = sc.broadcast(cars)

        val filterRDD = rdd.filter(line => {
          val split = line.split("\t")
          val car = split(3)
          broCars.value.contains(car)
        })


        //将结果保存到hbase
        filterRDD.foreachPartition(iter => {

          // create  't_monitor',{NAME => 'info' ,VERSIONS => 1000}

          //创建hbase连接
          val hbaseClient = HbaseUtil.getConnection
          val table = hbaseClient.getTable("t_monitor")


          iter.foreach(line => {
            val split = line.split("\t")

            val date = split(0)
            val monitor_id = split(1)
            val camera_id = split(2)
            val car = split(3)
            val action_time = split(4)
            val speed = split(5)
            val road_id = split(6)
            val area_id = split(7)


            val format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss")
            val ts = format.parse(action_time).getTime

            val rowkey = car + "_" + date

            val put = new Put(rowkey.getBytes())

            put.add("info".getBytes, "monitor_id".getBytes(), ts, monitor_id.getBytes)
            put.add("info".getBytes, "camera_id".getBytes(), ts, camera_id.getBytes)
            put.add("info".getBytes, "speed".getBytes(), ts, speed.getBytes)
            put.add("info".getBytes, "road_id".getBytes(), ts, road_id.getBytes)
            put.add("info".getBytes, "area_id".getBytes(), ts, area_id.getBytes)

            table.put(put)
          })

        })
      })


    //cars.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }

  /**
    * 初始化spark配置
    *  conf.setMaster("local")
    */
  override def init(): Unit = {
    conf.setMaster("local[2]")
  }
}
